//
// Created by QU on 24-9-5.
//

// 发送的格式改变了.

/*
 *因为TCP底层通信是面向字节流的，TCP只保证发送数据的准确性和顺序性，字节流以字节为单位，客户端每
 *次发送N个字节给服务端，N取决于当前客户端的发送缓冲区是否有数据，比如发送缓冲区总大小为10个字
 *节，当前有5个字节数据（上次要发送的数据比如’loveu＇)未发送完，那么此时只有5个字节空闲空间，我们
 *调用发送接口发送helloworld！其实就是只能发送Hello给服务器，那么服务器一次性读取到的数据就很可能
 *是loveuhello。而剩余的world！只能留给下一次发送，下一次服务器接收到的就是world！
 */


/* 这是最好理解的粘包问题的产生原因。还有一些其他的原因比如
 * 1 客户端的发送频率远高于服务器的接收频率，就会导致数据在服务器的tcp接收缓冲区滞留形成粘连，
         比如客户端1s内连续发送了两个helloworld！，服务器过了2s才接收数据，那一次性读出两个helloworld！。
 * 2 tcp底层的安全和效率机制不允许字节数特别少的小包发送频率过高，tcp会在底层累计数据长度到一定大小才一起发送，
         比如连续发送1字节的数据要累计到多个字节才发送，可以了解下tcp底层的Nagle算法。
 * 3 再就是我们提到的最简单的情况，发送端缓冲区有上次未发送完的数据或者接收端的缓冲区里有未取出的数据导致数据粘连。 
 */

/**
 * 处理粘包的方式主要采用应用层定义收发包格式的方式，这个过程俗称切包处理，常用的协议被称为`tlv`协议
 *  (消息id+消息长度+消息内容)
 * 为保证大家容易理解，我们先简化发送的格式，格式变为消息长度+消息内容的方式，之后再完善为tlv格式。
 */


/**
*async_write_some 和 async_send 都是 Boost.Asio 中用于异步写数据的函数，但它们在使用场景和底层实现上存在一些区别。下面从效率和实现的角度详细说明：

1. async_write_some 的特点：
底层控制较少：它更像是一个低级别的异步写操作，用于尽可能多地写入数据到缓冲区中，但不会尝试完成整个写入操作。它的行为类似于“部分写入”。
部分写入：调用后，它只会写入 socket 缓冲区中当前能够接收的数据。剩余数据需要在回调中继续处理。
效率：由于它不强制等待所有数据写完，传输效率较高，尤其适用于自定义协议或需要自己管理发送缓冲区的场景，减少了阻塞风险。
应用场景：适用于需要频繁发送小块数据，并且对每次写操作进行细粒度控制的场景。
2. async_send 的特点：
协议相关的发送：async_send 专门设计用于 TCP/UDP socket 通信，并且允许使用发送标志（flags）来控制发送行为，如 MSG_DONTWAIT、MSG_NOSIGNAL 等。
全数据发送：默认情况下，async_send 会尽量一次性发送所有数据，但可能还需要多次调用才能完成整个写入。
效率：因为它与协议相关，且支持一些额外的网络层标志，async_send 在某些特定的网络优化场景（如非阻塞或避免信号干扰）可能会更高效，但如果不使用这些标志，效率上的优势并不明显。
应用场景：适合需要使用底层 socket flags 的复杂场景，比如需要对发送行为做精细控制或优化底层 TCP 传输性能。
3. 效率差异分析：
最小写入单元：async_write_some 每次写入数据的量可能会更小（具体取决于系统缓冲区的可用容量），因此可能需要多次写入才能发送完所有数据。但正因为它是“部分写入”，它不会等待缓冲区完全有空，减少了潜在的等待时间，适合对传输性能要求高的场景。
系统调用开销：对于较大的数据传输，async_send 可能更具优势，因为它会尽量一次性发送尽可能多的数据，减少了系统调用的次数。
flags 的使用：async_send 的 flags（如 MSG_DONTWAIT）可以进一步优化特定场景下的性能，例如非阻塞发送模式下减少延迟。但在不需要这些特性的情况下，async_send 和 async_write_some 的性能差异可能非常小。
4. 使用建议：
async_write_some：适用于需要频繁发送小数据块，或者想要精确控制数据传输过程的场景，比如自定义网络协议或需要管理缓冲区。
async_send：适合希望使用发送标志优化传输，或需要控制协议细节的场景。如果你使用标准的网络协议（如 TCP、UDP）并且不需要处理每次写入的数据量，async_send 可以减少代码复杂度。
5. 总结：
效率比较：在常规数据发送中，async_send 和 async_write_some 的性能差异并不显著。如果你需要自定义网络层行为或优化协议细节，async_send 提供了更多的灵活性和潜在的优化空间。对于不需要这些特性的场景，async_write_some 更加简洁，并且它在处理小块数据的高频发送时，通常表现得更加高效。
*/

/*
Handling read: 6d31929f-1920-46e7-b579-eb2443ced714
Received bytes: 6d31929f-1920-46e7-b579-eb2443ced714
33
Content: 6d31929f-1920-46e7-b579-eb2443ced714
data_len is 31
receive data is: 000001E48EE54690
Welcome aquawius to asio world!
Handling write: 6d31929f-1920-46e7-b579-eb2443ced714
Sending message: 6d31929f-1920-46e7-b579-eb2443ced714
Handling read: 6d31929f-1920-46e7-b579-eb2443ced714
Received bytes: 6d31929f-1920-46e7-b579-eb2443ced714
66                                                          // sticky!!
Content: 6d31929f-1920-46e7-b579-eb2443ced714
data_len is 31
receive data is: 000001E48EE542D0
Welcome aquawius to asio world!
data_len is 31
receive data is: 000001E48EE54690
Welcome aquawius to asio world!
Handling write: 6d31929f-1920-46e7-b579-eb2443ced714
Sending message: 6d31929f-1920-46e7-b579-eb2443ced714
Handling read: 6d31929f-1920-46e7-b579-eb2443ced714
Received bytes: 6d31929f-1920-46e7-b579-eb2443ced714
33
Content: 6d31929f-1920-46e7-b579-eb2443ced714
data_len is 31
receive data is: 000001E48EE54810
Welcome aquawius to asio world!
Handling write: 6d31929f-1920-46e7-b579-eb2443ced714
Sending message: 6d31929f-1920-46e7-b579-eb2443ced714
 */

#include "08broken_my_tlv_pkg_sticky_tcp_server.h"


Session::Session(asio::io_context& io_context, Server& server) : server_(server), socket_(io_context)
{
    auto uuid = boost::uuids::random_generator()();
    uuid_ = boost::uuids::to_string(uuid);
    std::cout << "Session created with UUID: " << uuid_ << std::endl;

    recv_msg_node_ = std::make_shared<MsgNode>(MsgNode::MAX_LENGTH);
}

Session::~Session()
{
    std::cout << "Session destroyed: " << uuid_ << std::endl;
}

tcp::socket& Session::socket() { return socket_; }

void Session::start()
{
    std::cout << "Session started: " << uuid_ << std::endl;
    // memset()............
    // std::fill(std::begin(data_), std::end(data_), 0);            // c++11
    std::ranges::fill(data_, std::byte { 0 }); // c++20

    auto self { shared_from_this() }; // 获取一个自身的 shared_ptr
    socket_.async_read_some(asio::buffer(data_, Session::MAX_LENGTH),
                            [self, this](std::error_code error_code, std::size_t bytes_transferred)
                            {
                                HandlerRead(error_code, bytes_transferred);
                            }

                            // std::bind(&handler_read, this,  std::placeholders::_1, std::placeholders::_2, self)
    );
}

std::string& Session::getUuid()
{
    return uuid_;
}

void Session::Send(std::byte* message, size_t max_length)
{
    auto self { shared_from_this() }; // 获取一个自身的 shared_ptr
    std::shared_ptr<MsgNode> queue_first_node_p;

    {
        std::lock_guard lock_guard(send_mutex_);
        send_msg_queue_.emplace(make_shared<MsgNode>(message, max_length));

        // 前面emplace, 已经是1
        if (send_msg_queue_.size() > 1)
        {
            // 如果队列之前是空的，启动异步发送, 在队列中, 回调负责处理.
            return;
        }
        queue_first_node_p = send_msg_queue_.front();
    }

    std::cerr << "Sending message: " << uuid_ << std::endl;
    std::cerr << "Non-Handler Sending message Content: ";
    std::cerr << reinterpret_cast<char*>(data_) << std::endl;

    size_t head_data_content_length = 0;
    memcpy(&head_data_content_length, recv_msg_node_->data_, MsgNode::HEADER_LENGTH);
    std::cerr << "Sending message length: " << head_data_content_length << std::endl;

    socket_.async_send(
        asio::buffer(queue_first_node_p->data_, head_data_content_length),
        [self, this](const boost::system::error_code& error_code, size_t bytes_transferred)
        {
            HandlerWrite(error_code, bytes_transferred);
        }
    );
}

void Session::HandlerWrite(const boost::system::error_code& error_code, size_t bytes_transferred)
{
    auto self = shared_from_this();
    std::cout << "Handling write: " << uuid_ << std::endl;

    if (error_code)
    {
        std::cerr << "Handle an error: " << uuid_ << std::endl;
        std::cerr << error_code.value() << ". Message: " << error_code.message()
            << std::endl;
        server_.ClearSessionWithUUID(uuid_);
        std::cerr << "ClearSession: " << uuid_ << std::endl;
        return;
    }

    std::shared_ptr<MsgNode> queue_first_node_p;
    {
        std::lock_guard lock_guard(send_mutex_);

        // 移除已发送的消息
        if (!send_msg_queue_.empty())
        {
            send_msg_queue_.pop();
        }

        // 如果队列为空，直接返回，不再发送
        if (send_msg_queue_.empty())
        {
            return;
        }

        // 取出队列中的下一个消息
        queue_first_node_p = send_msg_queue_.front();
    }

    std::cerr << "Sending message: " << uuid_ << std::endl;
    std::cerr << "In-Handler Sending message Content: ";
    std::cerr << reinterpret_cast<char*>(data_) << std::endl;

    size_t head_data_content_length = 0;
    memcpy(&head_data_content_length, recv_msg_node_->data_, MsgNode::HEADER_LENGTH);
    std::cerr << "Sending message length: " << head_data_content_length << std::endl;

    socket_.async_send(
        asio::buffer(queue_first_node_p->data_, head_data_content_length),
        [self, this](const boost::system::error_code& error_code, size_t bytes_transferred)
        {
            HandlerWrite(error_code, bytes_transferred);
        }
    );
}

void Session::PrintBufferContent(const std::byte* data, const size_t size)
{
    for (size_t i = 0; i < size; ++i)
    {
        // 打印每个字节的数据，可以根据需求打印为十六进制或十进制
        std::cout << std::hex << std::to_integer<int>(data[i]);
    }
    std::cout << std::dec << std::endl; // 恢复到十进制输出
}

void Session::PrintBufferContentRed(const std::byte* data, const size_t size)
{
    for (size_t i = 0; i < size; ++i)
    {
        // 打印每个字节的数据，可以根据需求打印为十六进制或十进制
        std::cerr << std::hex << std::to_integer<int>(data[i]);
    }
    std::cerr << std::dec << std::endl; // 恢复到十进制输出
}

void Session::HandlerRead(const boost::system::error_code& error_code, size_t bytes_transferred)
{
    auto self { shared_from_this() }; // 获取一个自身的 shared_ptr

    // std::cout << "Handling read: " << uuid_ << std::endl;
    if (error_code)
    {
        if (error_code == asio::error::eof)
        {
            // std::cout << "Received EOF: " << uuid_ << std::endl;
            server_.ClearSessionWithUUID(uuid_);
            std::cerr << "ClearSession: " << uuid_ << std::endl;
            return;
        }
        std::cerr << "Handle an error: " << uuid_ << error_code.value() << ". Message: " << error_code.message() <<
            std::endl;
        // throw std::runtime_error("Received error.");
        server_.ClearSessionWithUUID(uuid_);
        std::cerr << "ClearSession: " << uuid_ << std::endl;
        return;
    }

    std::cout << "Received bytes: " << uuid_ << "\t\tbytes:" << bytes_transferred << std::endl;
    // std::cout << "RECEIVED orig content: ";
    // PrintBufferContent(data_, bytes_transferred);

    // std::ranges::fill(data_, data_ + MAX_LENGTH, 0); // c++20

    // 粘包处理测试代码
    using namespace std::chrono_literals; // for "ms" user custom literals.
    std::chrono::milliseconds duration(500ms);
    std::this_thread::sleep_for(duration);

    size_t copy_len = 0;
    while (bytes_transferred > 0)
    {
        // header_parse_flag 为 true 表示正在处理头, 初始值为false
        if (!header_parse_flag) // 没处理头
        {
            //收到的数据不足头部大小, 调用回调接着收, flag不变
            if (bytes_transferred + recv_msg_node_->cur_len_ < MsgNode::HEADER_LENGTH)
            {
                memcpy(recv_msg_node_->data_ + recv_msg_node_->cur_len_, data_ + copy_len, bytes_transferred);
                recv_msg_node_->cur_len_ += bytes_transferred;
                ::memset(data_, 0, MAX_LENGTH);
                socket_.async_read_some(boost::asio::buffer(data_, MAX_LENGTH),
                                        [self, this](const boost::system::error_code& ec, size_t bytes_transferred)
                                        {
                                            this->HandlerRead(ec, bytes_transferred);
                                        });
                return;
            }
            //收到的数据大于头部大小, 头是完整的, body可能超过也可能不超过
            //头部剩余未复制的长度, 就是身体长度
            size_t head_remain = MsgNode::HEADER_LENGTH - recv_msg_node_->cur_len_;
            memcpy(recv_msg_node_->data_ + recv_msg_node_->cur_len_, data_ + copy_len, head_remain);
            //更新已处理的data长度和剩余未处理的长度
            copy_len += head_remain;
            bytes_transferred -= head_remain;

            //获取头部包含的字节数据 (表示长度)
            size_t head_data_content_length = 0;
            memcpy(&head_data_content_length, recv_msg_node_->data_, MsgNode::HEADER_LENGTH);
            std::cout << "head_data_content_length is: " << head_data_content_length << std::endl;
            //头部长度非法
            if (head_data_content_length > MAX_LENGTH)
            {
                std::cout << "invalid data length is: " << head_data_content_length << std::endl;
                server_.ClearSessionWithUUID(uuid_);
                return;
            }
            recv_msg_node_ = std::make_shared<MsgNode>(head_data_content_length);

            //消息的长度小于头部规定的长度，说明数据未收全，则先将部分消息放到接收节点里
            if (bytes_transferred < head_data_content_length)
            {
                memcpy(recv_msg_node_->data_ + recv_msg_node_->cur_len_, data_ + copy_len, bytes_transferred);
                recv_msg_node_->cur_len_ += bytes_transferred;
                ::memset(data_, 0, MAX_LENGTH);
                socket_.async_read_some(boost::asio::buffer(data_, MAX_LENGTH),
                                        [self, this](const boost::system::error_code& ec, size_t bytes_transferred)
                                        {
                                            this->HandlerRead(ec, bytes_transferred);
                                        });;
                //头部处理完成
                header_parse_flag = true;
                return;
            }

            // 根据头部指示的包长度, 数据收全, 可能存在粘包中多出的数据, 需要在下一个包处理
            memcpy(recv_msg_node_->data_ + recv_msg_node_->cur_len_, data_ + copy_len, head_data_content_length);
            recv_msg_node_->cur_len_ += head_data_content_length;
            copy_len += head_data_content_length;
            bytes_transferred -= head_data_content_length;

            // 打印出接受完全的包
            recv_msg_node_->data_[recv_msg_node_->max_len_] = std::byte { 0 };
            std::cout << "receive data is: " << std::endl;
            std::cout << reinterpret_cast<char*>(recv_msg_node_->data_) << std::endl;
            std::cout << "receive data length is: " << recv_msg_node_->max_len_ << std::endl;

            //此处可以调用Send发送测试

            Send(recv_msg_node_->data_, recv_msg_node_->max_len_);
            // Send(recv_msg_node_->data_, recv_msg_node_->max_len_);

            //继续轮询剩余未处理数据, 进入下一个while循环, bytes_transferred已经减去上一个包的长度.
            header_parse_flag = false;
            recv_msg_node_->ClearMsg();
            if (bytes_transferred <= 0)
            {
                ::memset(data_, 0, MAX_LENGTH);
                socket_.async_read_some(boost::asio::buffer(data_, MAX_LENGTH),
                                        [self, this](const boost::system::error_code& ec, size_t bytes_transferred)
                                        {
                                            this->HandlerRead(ec, bytes_transferred);
                                        });
                return;
            }
            continue;
        }

        //已经处理完头部，处理上次未接受完的消息数据
        //接收的数据仍不足剩余未处理的
        size_t remain_msg = recv_msg_node_->max_len_ - recv_msg_node_->cur_len_;
        if (bytes_transferred < remain_msg)
        {
            memcpy(recv_msg_node_->data_ + recv_msg_node_->cur_len_, data_ + copy_len, bytes_transferred);
            recv_msg_node_->cur_len_ += bytes_transferred;
            ::memset(data_, 0, MAX_LENGTH);
            socket_.async_read_some(boost::asio::buffer(data_, MAX_LENGTH),
                                    [self, this](const boost::system::error_code& ec, size_t bytes_transferred)
                                    {
                                        this->HandlerRead(ec, bytes_transferred);
                                    }
            );
            return;
        }
        memcpy(recv_msg_node_->data_ + recv_msg_node_->cur_len_, data_ + copy_len, remain_msg);
        recv_msg_node_->cur_len_ += remain_msg;
        bytes_transferred -= remain_msg;
        copy_len += remain_msg;
        recv_msg_node_->data_[recv_msg_node_->max_len_] = std::byte { 0 };
        std::cout << "receive data is: " << std::endl;
        std::cout << reinterpret_cast<char*>(recv_msg_node_->data_) << std::endl;
        std::cout << "receive data length is: " << recv_msg_node_->max_len_ << std::endl;

        //此处可以调用Send发送测试
        Send(recv_msg_node_->data_, recv_msg_node_->max_len_);
        // Send(recv_msg_node_->data_, recv_msg_node_->max_len_);
        //继续轮询剩余未处理数据
        header_parse_flag = false;
        recv_msg_node_->ClearMsg();
        if (bytes_transferred <= 0)
        {
            ::memset(data_, 0, MAX_LENGTH);
            socket_.async_read_some(boost::asio::buffer(data_, MAX_LENGTH),
                                    [self, this](const boost::system::error_code& ec, size_t bytes_transferred)
                                    {
                                        this->HandlerRead(ec, bytes_transferred);
                                    }
            );
            return;
        }
        continue;
    }

    //
    // socket_.async_receive(asio::buffer(data_, bytes_transferred),
    //                       [self, this](const boost::system::error_code& error_code, std::size_t bytes_transferred)
    //                       {
    //                           this->handlerRead(error_code, bytes_transferred);
    //                       });
    //
    // 这里同样将receive的数据发回给client, 完成echo.
    // socket_.async_send(boost::asio::buffer(data_, bytes_transferred),
    //                    [self, this](const boost::system::error_code& ec, size_t bytes_transferred)
    //                    {
    //                        this->handlerWrite(ec, bytes_transferred);
    //                    }
    // );
}


Server::Server(asio::io_context& io_context, unsigned short port) :
    io_context_(io_context), acceptor_(io_context, tcp::endpoint(tcp::v4(), port))
{
    std::cout << "Server starting on port: " << port << std::endl;
    start_accept();
}

void Server::ClearSessionWithUUID(std::string const& uuid)
{
    auto n = session_map_.erase(uuid);
    if (n == 0)
    {
        std::cerr << "Session with UUID: " << uuid << " not exists." << std::endl;
    }
    else
    {
        std::cerr << "Session with UUID: " << uuid << " erase success." << std::endl;
    }
}


void Server::start_accept()
{
    std::cout << "Server accepting connections" << std::endl;
    auto new_session_p = std::make_shared<Session>(io_context_, *this);
    acceptor_.async_accept(new_session_p->socket(),
                           // 这里的捕获列表中不能使用 &, 会导致shared_ptr在这个异步调用中被析构.
                           [=, this](const boost::system::error_code& code)
                           {
                               handle_accept(new_session_p, code);
                           }
                           // std::bind(&handle_accept, this, )
    );
}

void Server::handle_accept(const std::shared_ptr<Session>& new_session_p, const boost::system::error_code& error_code)
{
    // accept error
    std::cout << "Handling accept" << std::endl;
    if (error_code)
    {
        std::cerr << "Handle error: " << error_code.value() << ". Message: " << error_code.message()
            << std::endl;
        return;
    }

    // accept success
    new_session_p->start();
    session_map_.insert({ new_session_p->getUuid(), new_session_p });
    start_accept();
}

MsgNode::MsgNode(size_t max_len)
{
    if (max_len > MAX_LENGTH)
    {
        throw std::length_error("Message exceeds max length");
    }

    max_len_ = max_len + HEADER_LENGTH;
    data_ = new std::byte[max_len_ + 1]; // 分配内存

    std::ranges::fill(data_, data_ + max_len_, std::byte { 0 });

    data_[max_len_] = std::byte { 0 }; // 设置结束符
}

MsgNode::MsgNode(const std::byte* msg, size_t max_len)
{
    if (max_len > MAX_LENGTH)
    {
        throw std::length_error("Message exceeds max length");
    }

    max_len_ = max_len + HEADER_LENGTH;
    data_ = new std::byte[max_len_ + 1]; // 分配内存

    // 拷贝消息内容
    // memcpy(data_, &max_len_, HEADER_LENGTH);
    // memcpy(data_ + HEADER_LENGTH, msg, max_len);

    // 现代c++方式.
    // 拷贝消息长度到头部（用std::copy代替memcpy）
    // std::copy(reinterpret_cast<const std::byte*>(&max_len_),
    //           reinterpret_cast<const std::byte*>(&max_len_) + HEADER_LENGTH, data_);
    // 拷贝消息体到data_（用std::copy代替memcpy）
    // std::copy(msg, msg + max_len, data_ + HEADER_LENGTH);

    // 拷贝消息长度到头部（用std::copy代替memcpy）
    std::copy_n(reinterpret_cast<const std::byte*>(&max_len_),
                HEADER_LENGTH, data_);
    // 拷贝消息体到data_（用std::copy代替memcpy）
    std::copy_n(msg, max_len, data_ + HEADER_LENGTH);

    data_[max_len_] = std::byte { 0 }; // 设置结束符
}

void MsgNode::ClearMsg()
{
    std::ranges::fill(data_, data_ + max_len_, std::byte { 0 }); // 修正填充范围
    cur_len_ = 0;
}

MsgNode::~MsgNode()
{
    delete[] data_; // 释放内存
}

int main(int argc, char* argv[])
{
    asio::io_context io_context;

    try
    {
        Server server(io_context, 10120);
        io_context.run();
    }
    catch (std::exception& e)
    {
        std::cerr << "Exception: " << e.what() << std::endl;
    }
}
